package cn.com.infcn.spider.io.output.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;

import cn.com.infcn.spider.domain.Doc;
import cn.com.infcn.spider.io.output.Output;
import cn.com.infcn.spider.util.MongoDao;

import com.alibaba.fastjson.JSONObject;
import com.mongodb.BasicDBObject;
import com.mongodb.DBObject;

/**
 * jdbc输出
 * 
 * @author Ansj
 *
 */
public class mongoOutput implements Output {

	private ConcurrentLinkedQueue<JSONObject> queue = null;

	private MongoDao mongoDao = null;

	private String tableName = null;

	private int batchSize = 0;

	public mongoOutput(JSONObject job, String jobName) throws Exception {
		tableName = job.getString("tableName");
		batchSize = job.getIntValue("batchSize");
		if (tableName == null) {
			throw new Exception("not set tableName! in "
					+ job.toJSONString().replace("\n", " "));
		}

		if (batchSize > 1) {
			queue = new ConcurrentLinkedQueue<>();
		}

		mongoDao = new MongoDao(job.getString("host"), job.getInteger("port"),
				job.getString("dbName"));
	}

	@Override
	public void save(Doc doc) {
		
		//判断是否有_longtime字段，如果没有则内置一个时间戳字段类型
		boolean flag = doc.getResult().containsKey("_longtime");
		if(!flag){
			doc.getResult().put("_longtime", new Date().getTime());
		}
		
		if (queue == null) {
			mongoDao.save(tableName, doc.getResult());
		} else {
			queue.add(doc.getResult());
			if (queue.size() >= batchSize) {
				flush();
			}
		}

	}

	@Override
	public synchronized void flush() {
		batchSave();
	}

	public synchronized void batchSave() {

		if (queue == null) {
			return;
		}

		JSONObject job = null;

		synchronized (queue) {

			List<DBObject> all = new ArrayList<>(queue.size());

			while ((job = queue.poll()) != null) {
				all.add(new BasicDBObject(job));
			}

			mongoDao.getCollection(tableName).insert(all);
		}

	}

	@Override
	public void shutdown() {
		flush();
		mongoDao.getClient().close();
	}

}
